package com.bky.mq.base;

import com.bky.util.ExecutorThreadFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;

import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

public abstract class BaseProducer<T> {
    protected final Logger LOG = LoggerFactory.getLogger("BaseProducer");
    @Autowired
    private KafkaTemplate kafkaTemplate;
    private Executor executors = Executors.newSingleThreadExecutor(new ExecutorThreadFactory("sendMsgPool"));
    /**
     * 同步消息发送
     *
     * @param mqInfo 消息体
     * @return true 成功 false 失敗
     */
    public boolean synSend(T mqInfo) {
        try {
            kafkaTemplate.send(buildMessageSubject(), getBusinessId(mqInfo), getMessage(mqInfo));
            return true;
        } catch (Exception e) {
            LOG.error(getExceptionMessage(), e);
            return false;
        }
    }

    /**
     * 异步消息发送
     *
     * @param mqInfo 消息体
     * @return true 成功 false 失敗
     */
    public boolean asynSend(T mqInfo) {
        try {
            executors.execute(new CreateMessage(mqInfo));
            return true;
        } catch (Exception e) {
            LOG.error(getExceptionMessage(), e);
            return false;
        }
    }

    /**
     * 统计消息
     */
    class CreateMessage implements Runnable {
        private T mqInfo;

        public CreateMessage(T mqInfo) {
            this.mqInfo = mqInfo;
        }

        @Override
        public void run() {
            try {
                kafkaTemplate.send(buildMessageSubject(), getBusinessId(mqInfo), getMessage(mqInfo));
            } catch (Exception e) {
                LOG.error("baseProducer send msg error:" + e);
            }
        }
    }
    /**
     * 获得消息体
     *
     * @return
     */
    protected abstract String getMessage(T message);


    /**
     * 设置主题
     *
     * @return
     */
    protected abstract String buildMessageSubject();

    /**
     * 设置异常提示语
     *
     * @return
     */
    protected abstract String getExceptionMessage();

    /**
     * 获得业务ID
     *
     * @return 默认返回UUID
     */
    protected abstract String getBusinessId(T message);
}
